Dubbo源码分析10 -Dubbo 服务引入

发表于 | 分类于 |

计划阅读调试下Dubbo的源码，结合官方源码分析Dubbo，自身再分析总结

本文对应的Dubbo 服务引入

源码分析

订阅

上一篇分析了Dubbo服务引入的整体流程。我发现一直没有看到DubboProtocol的部分。重新debug之后，发现漏了一些流程

首先先找到DubboProtocol#ref方法的堆栈

调用堆栈

梳理下它的调用链

RegistryProtocol#doRefer -> RegistryDirectory#subscribe -> FailbackRegistry#subscribe -> ZookeeperRegistry#doSubscribe ->

FailbackRegistry#notify -> FailbackRegistry#doNotify -> AbstractRegistry#notify -> RegistryDirectory#notify ->

RegistryDirectory#refreshInvoker -> RegistryDirectory#toInvokers -> Protocol的动态类以及一些Wrapper -> DubboProtocol#refer

调用链是比较复杂的，从registryProtocol的refer ，在经过RegistryDirectory以及其基类之间的调用，最终将其转化为invoker，再最后调用到DubboProtocol#refer做最后的处理。从调用链可以看出RegistryDirectory是作为字典的存在，用来保证注册中心的节点变动实时体现在Invoker上

回到RegistryProtocol#doRefer中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry); // 此处registry对应的是ZookeeperReigistry
    directory.setProtocol(protocol);
    // 设置directory

    // all attributes of REFER_KEY
    Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
    //
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
    if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
        // 注册到注册中心
        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                Constants.CHECK_KEY, String.valueOf(false)));
    }
    // 订阅privider， configurators, routers
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
            Constants.PROVIDERS_CATEGORY
                    + "," + Constants.CONFIGURATORS_CATEGORY
                    + "," + Constants.ROUTERS_CATEGORY));

    Invoker invoker = cluster.join(directory);
    ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
    return invoker;
}

进入directory#subscribe方法内

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
    public void subscribe(URL url) {
        setConsumerUrl(url);
        // 对应的zookeeperRegistry
        registry.subscribe(url, this);
    }
}

// ZookeeperRegistry对应的基类
public abstract class FailbackRegistry extends AbstractRegistry {
   @Override
    public void subscribe(URL url, NotifyListener listener) {
      super.subscribe(url, listener);
      removeFailedSubscribed(url, listener);
      try {
          // Sending a subscription request to the server side
          doSubscribe(url, listener);
      } catch (Exception e) {
          Throwable t = e;

          List<URL> urls = getCacheUrls(url);
          if (urls != null && !urls.isEmpty()) {
              notify(url, listener, urls);
              logger.error("Failed to subscribe " + url + ", Using cached list: " + urls + " from cache file: " + getUrl().getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/dubbo-registry-" + url.getHost() + ".cache") + ", cause: " + t.getMessage(), t);
          } else {
              // If the startup detection is opened, the Exception is thrown directly.
              boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                      && url.getParameter(Constants.CHECK_KEY, true);
              boolean skipFailback = t instanceof SkipFailbackWrapperException;
              if (check || skipFailback) {
                  if (skipFailback) {
                      t = t.getCause();
                  }
                  throw new IllegalStateException("Failed to subscribe " + url + ", cause: " + t.getMessage(), t);
              } else {
                  logger.error("Failed to subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
              }
          }

          // Record a failed registration request to a failed list, retry regularly
          addFailedSubscribed(url, listener);
      }
  }

  // 省去无关代码
}

public class ZookeeperRegistry extends FailbackRegistry {
  @Override
  protected void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        // 当interface 参数为 * 时
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            String root = toRootPath();
            ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
            if (listeners == null) {
                zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                listeners = zkListeners.get(url);
            }
            ChildListener zkListener = listeners.get(listener);
            if (zkListener == null) {
                listeners.putIfAbsent(listener, new ChildListener() {
                    @Override
                    public void childChanged(String parentPath, List<String> currentChilds) {
                        for (String child : currentChilds) {
                            child = URL.decode(child);
                            if (!anyServices.contains(child)) {
                                anyServices.add(child);
                                subscribe(url.setPath(child).addParameters(Constants.INTERFACE_KEY, child,
                                        Constants.CHECK_KEY, String.valueOf(false)), listener);
                            }
                        }
                    }
                });
                zkListener = listeners.get(listener);
            }
            zkClient.create(root, false);
            List<String> services = zkClient.addChildListener(root, zkListener);
            if (services != null && !services.isEmpty()) {
                for (String service : services) {
                    service = URL.decode(service);
                    anyServices.add(service);
                    subscribe(url.setPath(service).addParameters(Constants.INTERFACE_KEY, service,
                            Constants.CHECK_KEY, String.valueOf(false)), listener);
                }
            }
        } else {
          // interface 参数有值时
            List<URL> urls = new ArrayList<URL>();
            // 按照分类 转换路径
            for (String path : toCategoriesPath(url)) {
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    listeners.putIfAbsent(listener, new ChildListener() {
                        @Override
                        public void childChanged(String parentPath, List<String> currentChilds) {
                            ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                        }
                    });
                    zkListener = listeners.get(listener);
                }
                // 此处zkClient为CuratorZookeeperClient
                // 创建非临时节点
                zkClient.create(path, false);
                List<String> children = zkClient.addChildListener(path, zkListener);
                if (children != null) {
                    urls.addAll(toUrlsWithEmpty(url, path, children));
                }
            }
            // 通知 此处调用父类方法
            notify(url, listener, urls);
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
  }



}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
public abstract class FailbackRegistry extends AbstractRegistry {
    @Override
    protected void notify(URL url, NotifyListener listener, List<URL> urls) {
        if (url == null) {
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("notify listener == null");
        }
        try {
           // 调用即类方法 
            doNotify(url, listener, urls);
        } catch (Exception t) {
            // Record a failed registration request to a failed list, retry regularly
            Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
            if (listeners == null) {
                failedNotified.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, List<URL>>());
                listeners = failedNotified.get(url);
            }
            listeners.put(listener, urls);
            logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
        }
    }
}

public abstract class AbstractRegistry implements Registry {
    protected void notify(URL url, NotifyListener listener, List<URL> urls) {
      // null 判断
        if (url == null) {
            throw new IllegalArgumentException("notify url == null");
        }
        if (listener == null) {
            throw new IllegalArgumentException("notify listener == null");
        }
        if ((urls == null || urls.isEmpty())
                && !Constants.ANY_VALUE.equals(url.getServiceInterface())) {
            logger.warn("Ignore empty notify urls for subscribe url " + url);
            return;
        }
        if (logger.isInfoEnabled()) {
            logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
        }

        // 按照 category
        Map<String, List<URL>> result = new HashMap<String, List<URL>>();
        for (URL u : urls) {
          // 此处判断逻辑为：
          // 1、 对比interface：consumerInterface 不为 * 并且 consumeInterface和providerInterface不一致
          // 2、 对比category： consumerCategory是否包含*; consumerCategory是否剔除(-)provider的category; consumerCategory是否包含providerCategory
          // 3、 providerUrl, consumerUrl中 enable字段为false
          // 下面是分别对group, version, classifier的判断。
          // 4、 group 对应为 consumerGroup为* 或者 consumer，provider一致或者 consumerGroup包含 providerGroup
          // 5、 version 对应为 consumerVersion 为* 或者 两者相等
          // 6、 classfier对应为 consumerClassifier 为null 或者 consumerClassifier 为* 或者 两者相等
          // 此处url为consumer协议的url
            if (UrlUtils.isMatch(url, u)) {
                String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
                List<URL> categoryList = result.get(category);
                if (categoryList == null) {
                    categoryList = new ArrayList<URL>();
                    result.put(category, categoryList);
                }
                categoryList.add(u);
            }
        }

        // result中是按照category来分类的

        if (result.size() == 0) {
            return;
        }


        Map<String, List<URL>> categoryNotified = notified.get(url);
        if (categoryNotified == null) {
            notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
            categoryNotified = notified.get(url);
        }
        for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
            String category = entry.getKey();
            List<URL> categoryList = entry.getValue();
            categoryNotified.put(category, categoryList);
            // 保存配置
            saveProperties(url);
            // 进行通知，此处调用回到了RegistryDirectory中
            listener.notify(categoryList);
        }
    }
}


public class RegistryDirectory<T> extends AbstractDirectory<T> implements NotifyListener {
    @Override
    public synchronized void notify(List<URL> urls) {
        List<URL> invokerUrls = new ArrayList<URL>();
        List<URL> routerUrls = new ArrayList<URL>();
        List<URL> configuratorUrls = new ArrayList<URL>();
        // 对url进行分类
        for (URL url : urls) {
            String protocol = url.getProtocol();
            String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            if (Constants.ROUTERS_CATEGORY.equals(category)
                    || Constants.ROUTE_PROTOCOL.equals(protocol)) {
                routerUrls.add(url);
            } else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
                    || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
                configuratorUrls.add(url);
            } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
                invokerUrls.add(url);
            } else {
                logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
            }
        }
        // configurators
        if (configuratorUrls != null && !configuratorUrls.isEmpty()) {
            // 转化为configurators对象列表
            this.configurators = toConfigurators(configuratorUrls);
        }
        // routers
        if (routerUrls != null && !routerUrls.isEmpty()) {
            // 转化为router列表
            List<Router> routers = toRouters(routerUrls);
            if (routers != null) { // null - do nothing
                setRouters(routers);
            }
        }
        List<Configurator> localConfigurators = this.configurators; // local reference
        // merge override parameters
        this.overrideDirectoryUrl = directoryUrl;
        if (localConfigurators != null && !localConfigurators.isEmpty()) {
            for (Configurator configurator : localConfigurators) {
                this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
            }
        }
        // providers
        // 针对provider刷新invokers，保证注册中心和字典内一致
        refreshInvoker(invokerUrls);
    }
}

TODO refresh invokers 等后续流程